package br.pucrs.lognplay.hadoop.test;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;

import br.pucrs.lognplay.hadoop.ElasticSearchClient;
import br.pucrs.lognplay.hadoop.HDFSTools;

public class Indexer extends Configured implements Tool {

	private static final Logger logger = Logger.getLogger(Indexer.class);

	/**
	 * args[0] index args[1] type args[2] client args[3] logSource
	 * 
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		args = new String[4];

		args[0] = "lognplay";
		args[1] = "logmsg";
		args[2] = "client-ADP";
		args[3] = "logSource-Odyssey1";

		logger.trace("Initializing ElasticSearchIndexer.....");
		int exitCode = ToolRunner.run(new Configuration(), new Indexer(), args);
		System.exit(exitCode);
	}

	public int run(String[] args) throws Exception {
		String index = args[0];
		String client = args[1];
		String type = args[2];
		String logSource = args[3];
		return run(index, client, type, logSource);
	}

	public int run(String index, String client, String type, String logSource) throws Exception {
		String inputPath = "/usr/lognplay/";
		String outputPath = "/usr/lognplay/output";

		JobConf conf = new JobConf(getConf(), Indexer.class);

		conf.setJobName("ElasticsearchIndexer");
		conf.setSpeculativeExecution(false);

		conf.set("index", index);
		conf.set("client", client);
		conf.set("type", type);
		conf.set("logSource", logSource);
		conf.set("inputPath", inputPath);
		conf.set("outputPath", outputPath);

		FileInputFormat.setInputPaths(conf, new Path(inputPath));
		FileOutputFormat.setOutputPath(conf, new Path(outputPath));

		conf.setInputFormat(TextInputFormat.class);
		conf.setMapperClass(Map.class);
		conf.setMapOutputKeyClass(NullWritable.class);
		conf.setMapOutputValueClass(NullWritable.class);
		conf.setNumReduceTasks(0);

		JobClient.runJob(conf);

		return 0;
	}

	public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, NullWritable, NullWritable> {

		private String line = null;

		private int linesProcessed = 0;

		private ElasticSearchClient esClient;

		private JobConf job;

		private static final Logger logger = Logger.getLogger(Indexer.class);

		public void configure(JobConf job) {
			try {
				this.job = job;

				logger.trace("========= Preparing to run MapJob ==========");
				logger.trace("JobName: " + job.getJobName());
				logger.trace("Params (");
				logger.trace("indexName: " + job.get("indexName", "lognplay"));
				logger.trace("mapType: " + job.get("mapType", "logmsg"));
				logger.trace("buffer: " + job.getInt("buffer", 100));
				logger.trace("ttl: " + job.getInt("ttl", 0));

				logger.trace(")");
				logger.trace("============================================");

				esClient = new ElasticSearchClient();
				esClient.startClient();
				esClient.setBufferSize(job.getInt("buffer", 100));
				esClient.setIndexName(job.get("indexName", "lognplay"));
				esClient.setType(job.get("mapType", "logmsg"));
				esClient.setTtl(job.getInt("ttl", 0));

			} catch (Exception e) {
				logger.error(e);
				e.printStackTrace();
			}
		}

		public void map(LongWritable key, Text value, OutputCollector<NullWritable, NullWritable> output, Reporter reporter) throws IOException {
			line = value.toString();
			try {
				linesProcessed++;
				esClient.put(line);
			} catch (Exception e) {
				logger.error(e);
				e.printStackTrace();
			}
		}

		public void close() throws IOException {
			try {
				logger.trace("============= Report ==============");
				logger.trace("JobName: " + job.getJobName());

				logger.trace("Lines Processed: " + linesProcessed);
				logger.trace("====================================");

				esClient.stopClient();

				HDFSTools.rm(job.get("outputPath"));
				HDFSTools.getFileSystem().close();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}

	}

}
